package com.example.canalclientsync.script;

import com.google.common.collect.Lists;
import com.example.canalclientsync.utils.CommonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author chen.cr
 * @since 2023/10/17
 */
@Slf4j
public abstract class BasicSync<T, B> {

    // 批量查库
    private static final int LIMIT = CommonUtils.BATCH_SIZE;

    @Resource
    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 本地简单测试用
     */
    @Value("${run.simple-test:false}")
    private Boolean runLocalTest;

    // 获取总数
    abstract long countTotal();

    // 分页批量查询
    abstract List<T> findPageData(long offset, int limit);

    // 数据同步逻辑
    abstract void sync0(T t);

    // get key id
    abstract B getKey(T t);


    /**
     * 数据同步
     *
     * @throws InterruptedException
     */
    public void syncData() throws InterruptedException {

        Long dataCount = countTotal();
        long iterations = (dataCount + LIMIT - 1) / LIMIT;
        log.info("------------- 总数据量：{}, 迭代次数：{}", dataCount, iterations);

        int size = dataCount.intValue();

        if (runLocalTest != null && runLocalTest) {
            // 仅测试2000条数据
            log.warn("Run simple test...............");
            iterations = 1;
            size = Long.valueOf(LIMIT * iterations).intValue();
        }

        final CountDownLatch LATCH = new CountDownLatch(size);

        // 收集异常数据进行重试
        final List<T> errorList = Lists.newArrayListWithExpectedSize(100);
        // 计数
        AtomicInteger idx = new AtomicInteger(0);

        for (long i = 0; i < iterations; i++) {
            // 批量查询mysql数据，进行同步
            long offset = i * LIMIT;
            List<T> pageData = findPageData(offset, LIMIT);

            for (T t : pageData) {
                // 异步进行
                threadPoolExecutor.execute(() -> {
                    try {
                        sync0(t);
                        log.info("OK. Now:{}, Total:{}", idx.incrementAndGet(), dataCount);

                    } catch (Exception e) {
                        log.error("Error, retry. modelId:{}. err:{}", getKey(t), e.getMessage());
                        errorList.add(t);

                    } finally {
                        LATCH.countDown();
                    }
                });
            }
        }

        LATCH.await();
        log.info("------------- 批量同步完成个数：{}，异常个数:{}", size, errorList.size());

        if (!CollectionUtils.isEmpty(errorList)) {
            log.info("错误数据重试开始\n\n");

            retryErrors(errorList, new AtomicInteger(1));
            log.info("异常数据全部重试结束\n\n");
        }

        log.info("------------- 模型历史数据同步结束！");
    }


    /**
     * 失败重试
     *
     * @param errorList  失败数据
     * @param retryCount 重试次数
     * @throws InterruptedException
     */
    private void retryErrors(Collection<T> errorList, AtomicInteger retryCount) throws InterruptedException {
        if (CollectionUtils.isEmpty(errorList)) {
            return;
        }

        int size = errorList.size();
        if (retryCount.get() > CommonUtils.RETRY_LIMIT) {
            String string = errorList.stream().map(this::getKey)
                    .map(String::valueOf).collect(Collectors.joining(",\n"));
            log.error("重试次数超限，重试结束。当前仍未成功同步的数据有：\n{}", string);
            return;
        }

        // 休息片刻
        Thread.sleep(5000);

        log.info("重试次数：{}，本轮重试数量：{}", retryCount.get(), size);

        final List<T> newErrors = Lists.newArrayListWithExpectedSize(size / 5);

        final CountDownLatch LATCH = new CountDownLatch(size);

        AtomicInteger idx = new AtomicInteger(0);

        for (T t : errorList) {
            threadPoolExecutor.execute(() -> {
                try {
                    sync0(t);
                    log.info("Retry OK. Now:{}, retry total:{}", idx.incrementAndGet(), size);

                } catch (Exception e) {
                    log.error("Retry Error. modelId:{}. err:{}", getKey(t), e.getMessage());
                    newErrors.add(t);

                } finally {
                    LATCH.countDown();
                }
            });
        }

        LATCH.await();
        log.info("第{}轮重试完成", retryCount.getAndIncrement());

        // 重试
        retryErrors(newErrors, retryCount);

    }

}
